
<!DOCTYPE html>
<html>
	<head>
		<meta http-equiv="Content-Type" content="text/html; charset=utf-8">
		<title>actor: Go Coverage Report</title>
		<style>
			body {
				background: black;
				color: rgb(80, 80, 80);
			}
			body, pre, #legend span {
				font-family: Menlo, monospace;
				font-weight: bold;
			}
			#topbar {
				background: black;
				position: fixed;
				top: 0; left: 0; right: 0;
				height: 42px;
				border-bottom: 1px solid rgb(80, 80, 80);
			}
			#content {
				margin-top: 50px;
			}
			#nav, #legend {
				float: left;
				margin-left: 10px;
			}
			#legend {
				margin-top: 12px;
			}
			#nav {
				margin-top: 10px;
			}
			#legend span {
				margin: 0 5px;
			}
			.cov0 { color: rgb(192, 0, 0) }
.cov1 { color: rgb(128, 128, 128) }
.cov2 { color: rgb(116, 140, 131) }
.cov3 { color: rgb(104, 152, 134) }
.cov4 { color: rgb(92, 164, 137) }
.cov5 { color: rgb(80, 176, 140) }
.cov6 { color: rgb(68, 188, 143) }
.cov7 { color: rgb(56, 200, 146) }
.cov8 { color: rgb(44, 212, 149) }
.cov9 { color: rgb(32, 224, 152) }
.cov10 { color: rgb(20, 236, 155) }

		</style>
	</head>
	<body>
		<div id="topbar">
			<div id="nav">
				<select id="files">
				
				<option value="file0">github.com/louloulin/dataflare/pkg/plugin/wasm/actor/wasm_plugin_actor.go (78.1%)</option>
				
				<option value="file1">github.com/louloulin/dataflare/pkg/plugin/wasm/actor/wasm_plugin_manager_actor.go (58.8%)</option>
				
				</select>
			</div>
			<div id="legend">
				<span>not tracked</span>
			
				<span class="cov0">not covered</span>
				<span class="cov8">covered</span>
			
			</div>
		</div>
		<div id="content">
		
		<pre class="file" id="file0" style="display: none">package actor

import (
        "context"
        "encoding/json"
        "fmt"
        "time"

        "github.com/louloulin/dataflare/pkg/actor/lifecycle"
        "github.com/louloulin/dataflare/pkg/message"
        "github.com/louloulin/dataflare/pkg/plugin"
)

// WasmPluginMessages 定义了WASM插件Actor的消息类型
type WasmPluginMessages struct{}

// ProcessDataMessage 处理数据消息
type ProcessDataMessage struct {
        Data *message.BaseMessage
}

// InitPluginMessage 初始化插件消息
type InitPluginMessage struct {
        Config map[string]interface{}
}

// GetInfoMessage 获取插件信息消息
type GetInfoMessage struct{}

// GetInfoResponse 获取插件信息响应
type GetInfoResponse struct {
        Info plugin.PluginInfo
}

// ProcessResponse 处理响应
type ProcessResponse struct {
        Result *message.BaseMessage
        Error  error
}

// WasmPluginActor 表示一个WASM插件Actor
type WasmPluginActor struct {
        lifecycle.LifecycleAwareActor
        plugin          plugin.Plugin
        processor       plugin.ProcessorPlugin
        lastProcessTime time.Time
        processCount    uint64
        processDuration time.Duration
}

// NewWasmPluginActor 创建一个新的WASM插件Actor
func NewWasmPluginActor(wasmPlugin plugin.Plugin) *WasmPluginActor <span class="cov8" title="1">{
        actor := &amp;WasmPluginActor{
                plugin:       wasmPlugin,
                processCount: 0,
        }

        // 尝试将插件转换为处理器类型
        if proc, ok := wasmPlugin.(plugin.ProcessorPlugin); ok </span><span class="cov8" title="1">{
                actor.processor = proc
        }</span>

        // 初始化生命周期相关属性
        <span class="cov8" title="1">actor.Initialize(wasmPlugin.GetID())

        return actor</span>
}

// OnInit 初始化插件
func (a *WasmPluginActor) OnInit() error <span class="cov8" title="1">{
        // 调用基类方法
        if err := a.LifecycleAwareActor.OnInit(); err != nil </span><span class="cov0" title="0">{
                return err
        }</span>

        <span class="cov8" title="1">return nil</span>
}

// OnStart 启动插件
func (a *WasmPluginActor) OnStart() error <span class="cov8" title="1">{
        // 调用基类方法
        if err := a.LifecycleAwareActor.OnStart(); err != nil </span><span class="cov0" title="0">{
                return err
        }</span>

        // 启动WASM插件
        <span class="cov8" title="1">ctx := context.Background()
        if err := a.plugin.Start(ctx); err != nil </span><span class="cov0" title="0">{
                a.RecordError(err)
                return fmt.Errorf("failed to start WASM plugin: %w", err)
        }</span>

        <span class="cov8" title="1">return nil</span>
}

// OnStop 停止插件
func (a *WasmPluginActor) OnStop() error <span class="cov8" title="1">{
        // 调用基类方法
        if err := a.LifecycleAwareActor.OnStop(); err != nil </span><span class="cov0" title="0">{
                return err
        }</span>

        // 停止WASM插件
        <span class="cov8" title="1">ctx := context.Background()
        if err := a.plugin.Stop(ctx); err != nil </span><span class="cov0" title="0">{
                a.RecordError(err)
                return fmt.Errorf("failed to stop WASM plugin: %w", err)
        }</span>

        <span class="cov8" title="1">return nil</span>
}

// HandleProcessMessage 处理数据消息
func (a *WasmPluginActor) HandleProcessMessage(msg *ProcessDataMessage) (*ProcessResponse, error) <span class="cov8" title="1">{
        if !a.IsRunning() </span><span class="cov8" title="1">{
                return nil, fmt.Errorf("plugin is not running")
        }</span>

        <span class="cov8" title="1">if a.processor == nil </span><span class="cov0" title="0">{
                return nil, fmt.Errorf("plugin does not implement ProcessorPlugin interface")
        }</span>

        // 处理消息
        <span class="cov8" title="1">a.RecordMessageReceived()
        startTime := time.Now()

        ctx := context.Background()
        result, err := a.processor.Process(ctx, msg.Data)

        // 计算处理时间
        processTime := time.Since(startTime)
        a.processDuration += processTime
        a.lastProcessTime = time.Now()
        a.processCount++

        if err != nil </span><span class="cov0" title="0">{
                a.RecordError(err)
                return &amp;ProcessResponse{
                        Error: fmt.Errorf("failed to process message: %w", err),
                }, nil
        }</span>

        <span class="cov8" title="1">a.RecordMessageSent()
        return &amp;ProcessResponse{
                Result: result.(*message.BaseMessage),
        }, nil</span>
}

// HandleInitMessage 处理初始化消息
func (a *WasmPluginActor) HandleInitMessage(msg *InitPluginMessage) error <span class="cov8" title="1">{
        if a.IsRunning() </span><span class="cov0" title="0">{
                return fmt.Errorf("plugin is already initialized")
        }</span>

        // 初始化插件
        <span class="cov8" title="1">ctx := context.Background()
        if err := a.plugin.Initialize(ctx, msg.Config); err != nil </span><span class="cov8" title="1">{
                a.RecordError(err)
                return fmt.Errorf("failed to initialize plugin: %w", err)
        }</span>

        <span class="cov8" title="1">return nil</span>
}

// HandleGetInfoMessage 处理获取信息消息
func (a *WasmPluginActor) HandleGetInfoMessage(msg *GetInfoMessage) (*GetInfoResponse, error) <span class="cov8" title="1">{
        info := a.plugin.GetInfo()

        // 扩展信息包含处理统计
        infoMap := map[string]interface{}{
                "id":            info.ID,
                "name":          info.Name,
                "type":          info.Type,
                "version":       info.Version,
                "api_version":   info.APIVersion,
                "description":   info.Description,
                "process_count": a.processCount,
        }

        if a.processCount &gt; 0 </span><span class="cov0" title="0">{
                infoMap["avg_process_time_ms"] = float64(a.processDuration.Milliseconds()) / float64(a.processCount)
                infoMap["last_process_time"] = a.lastProcessTime.Format(time.RFC3339)
        }</span>

        // 扩展插件信息
        <span class="cov8" title="1">extendedInfo := plugin.PluginInfo{
                ID:          info.ID,
                Name:        info.Name,
                Type:        info.Type,
                Mode:        info.Mode,
                Version:     info.Version,
                APIVersion:  info.APIVersion,
                Author:      info.Author,
                Description: info.Description,
                Config:      infoMap,
        }

        return &amp;GetInfoResponse{
                Info: extendedInfo,
        }, nil</span>
}

// GetMetrics 获取插件度量指标
func (a *WasmPluginActor) GetMetrics() map[string]interface{} <span class="cov8" title="1">{
        metrics := map[string]interface{}{
                "id":                a.plugin.GetID(),
                "type":              a.plugin.GetInfo().Type,
                "state":             string(a.State),
                "process_count":     a.processCount,
                "messages_received": a.Metrics.MessagesReceived,
                "messages_sent":     a.Metrics.MessagesSent,
                "error_count":       a.Metrics.ErrorCount,
        }

        if a.Metrics.ErrorCount &gt; 0 </span><span class="cov0" title="0">{
                metrics["last_error"] = a.Metrics.LastError
        }</span>

        <span class="cov8" title="1">if a.processCount &gt; 0 </span><span class="cov8" title="1">{
                metrics["avg_process_time_ms"] = float64(a.processDuration.Milliseconds()) / float64(a.processCount)
                metrics["last_process_time"] = a.lastProcessTime.Format(time.RFC3339)
        }</span>

        <span class="cov8" title="1">return metrics</span>
}

// String 返回Actor的字符串表示
func (a *WasmPluginActor) String() string <span class="cov8" title="1">{
        info := a.plugin.GetInfo()
        metrics := a.GetMetrics()

        metricsJson, _ := json.MarshalIndent(metrics, "", "  ")
        return fmt.Sprintf("WasmPluginActor[id=%s, name=%s, type=%s, state=%s]\nMetrics:\n%s",
                info.ID, info.Name, info.Type, a.State, string(metricsJson))
}</span>
</pre>
		
		<pre class="file" id="file1" style="display: none">package actor

import (
        "context"
        "fmt"
        "sync"
        "time"

        "github.com/louloulin/dataflare/pkg/actor/lifecycle"
        "github.com/louloulin/dataflare/pkg/plugin"
)

// WasmPluginManagerMessages 定义了WASM插件管理器Actor的消息类型
type WasmPluginManagerMessages struct{}

// LoadPluginMessage 加载插件消息
type LoadPluginMessage struct {
        PluginID        string
        PluginDirectory string
        Config          map[string]interface{}
}

// UnloadPluginMessage 卸载插件消息
type UnloadPluginMessage struct {
        PluginID string
}

// ListPluginsMessage 列出插件消息
type ListPluginsMessage struct{}

// ListPluginsResponse 列出插件响应
type ListPluginsResponse struct {
        Plugins []plugin.PluginInfo
}

// GetPluginMessage 获取插件消息
type GetPluginMessage struct {
        PluginID string
}

// GetPluginResponse 获取插件响应
type GetPluginResponse struct {
        Plugin     plugin.Plugin
        PluginInfo plugin.PluginInfo
        Actor      *WasmPluginActor
        Found      bool
}

// WasmPluginManagerActor 表示一个WASM插件管理器Actor
type WasmPluginManagerActor struct {
        lifecycle.LifecycleAwareActor
        plugins      map[string]plugin.Plugin
        pluginActors map[string]*WasmPluginActor
        pluginLoader *plugin.WasmPluginLoader
        pluginDir    string
        lastLoadTime time.Time
        mu           sync.RWMutex
}

// NewWasmPluginManagerActor 创建一个新的WASM插件管理器Actor
func NewWasmPluginManagerActor(pluginDir string) *WasmPluginManagerActor <span class="cov8" title="1">{
        actor := &amp;WasmPluginManagerActor{
                plugins:      make(map[string]plugin.Plugin),
                pluginActors: make(map[string]*WasmPluginActor),
                pluginDir:    pluginDir,
                pluginLoader: plugin.NewWasmPluginLoader(pluginDir),
        }

        // 初始化生命周期相关属性
        actor.Initialize("wasm_plugin_manager")

        return actor
}</span>

// OnInit 初始化插件管理器
func (a *WasmPluginManagerActor) OnInit() error <span class="cov8" title="1">{
        // 调用基类方法
        if err := a.LifecycleAwareActor.OnInit(); err != nil </span><span class="cov0" title="0">{
                return err
        }</span>

        // 扫描插件目录
        <span class="cov8" title="1">return nil</span>
}

// OnStart 启动插件管理器
func (a *WasmPluginManagerActor) OnStart() error <span class="cov8" title="1">{
        // 调用基类方法
        if err := a.LifecycleAwareActor.OnStart(); err != nil </span><span class="cov0" title="0">{
                return err
        }</span>

        // 可以在此添加额外的启动逻辑
        <span class="cov8" title="1">return nil</span>
}

// OnStop 停止插件管理器
func (a *WasmPluginManagerActor) OnStop() error <span class="cov8" title="1">{
        // 调用基类方法
        if err := a.LifecycleAwareActor.OnStop(); err != nil </span><span class="cov0" title="0">{
                return err
        }</span>

        // 停止所有插件
        <span class="cov8" title="1">a.mu.Lock()
        defer a.mu.Unlock()

        for _, pluginActor := range a.pluginActors </span><span class="cov8" title="1">{
                if err := pluginActor.OnStop(); err != nil </span><span class="cov0" title="0">{
                        a.RecordError(err)
                        // 继续停止其他插件，不中断
                }</span>
        }

        <span class="cov8" title="1">return nil</span>
}

// HandleLoadPlugin 处理加载插件消息
func (a *WasmPluginManagerActor) HandleLoadPlugin(msg *LoadPluginMessage) error <span class="cov8" title="1">{
        a.mu.Lock()
        defer a.mu.Unlock()

        a.RecordMessageReceived()

        // 如果指定了插件ID，尝试精确加载
        if msg.PluginID != "" </span><span class="cov8" title="1">{
                // 检查插件是否已加载
                if _, exists := a.plugins[msg.PluginID]; exists </span><span class="cov8" title="1">{
                        return fmt.Errorf("plugin %s already loaded", msg.PluginID)
                }</span>

                // 加载指定的插件
                <span class="cov8" title="1">pluginDir := a.pluginDir
                if msg.PluginDirectory != "" </span><span class="cov8" title="1">{
                        pluginDir = msg.PluginDirectory
                }</span>

                <span class="cov8" title="1">loader := plugin.NewWasmPluginLoader(pluginDir)
                ctx := context.Background()
                plugins, err := loader.LoadWasmPlugins(ctx)
                if err != nil </span><span class="cov0" title="0">{
                        a.RecordError(err)
                        return fmt.Errorf("failed to load plugin %s: %w", msg.PluginID, err)
                }</span>

                // 尝试从加载的插件中找到我们需要的
                <span class="cov8" title="1">var pluginInstance plugin.Plugin
                for _, p := range plugins </span><span class="cov0" title="0">{
                        if p.GetID() == msg.PluginID </span><span class="cov0" title="0">{
                                pluginInstance = p
                                break</span>
                        }
                }

                <span class="cov8" title="1">if pluginInstance == nil </span><span class="cov8" title="1">{
                        return fmt.Errorf("plugin %s not found in directory %s", msg.PluginID, pluginDir)
                }</span>

                // 创建插件Actor
                <span class="cov0" title="0">pluginActor := NewWasmPluginActor(pluginInstance)

                // 初始化插件
                if msg.Config != nil </span><span class="cov0" title="0">{
                        if err := pluginActor.HandleInitMessage(&amp;InitPluginMessage{Config: msg.Config}); err != nil </span><span class="cov0" title="0">{
                                a.RecordError(err)
                                return fmt.Errorf("failed to initialize plugin %s: %w", msg.PluginID, err)
                        }</span>
                }

                // 启动插件
                <span class="cov0" title="0">if err := pluginActor.OnStart(); err != nil </span><span class="cov0" title="0">{
                        a.RecordError(err)
                        return fmt.Errorf("failed to start plugin %s: %w", msg.PluginID, err)
                }</span>

                // 添加到管理的插件列表
                <span class="cov0" title="0">a.plugins[msg.PluginID] = pluginInstance
                a.pluginActors[msg.PluginID] = pluginActor

                a.lastLoadTime = time.Now()
                return nil</span>
        }

        // 否则，加载目录中的所有插件
        <span class="cov0" title="0">pluginDir := a.pluginDir
        if msg.PluginDirectory != "" </span><span class="cov0" title="0">{
                pluginDir = msg.PluginDirectory
        }</span>

        <span class="cov0" title="0">loader := plugin.NewWasmPluginLoader(pluginDir)
        ctx := context.Background()
        plugins, err := loader.LoadWasmPlugins(ctx)
        if err != nil </span><span class="cov0" title="0">{
                a.RecordError(err)
                return fmt.Errorf("failed to load plugins from directory %s: %w", pluginDir, err)
        }</span>

        <span class="cov0" title="0">for _, pluginInstance := range plugins </span><span class="cov0" title="0">{
                // 跳过已加载的插件
                pluginID := pluginInstance.GetID()
                if _, exists := a.plugins[pluginID]; exists </span><span class="cov0" title="0">{
                        continue</span>
                }

                // 创建插件Actor
                <span class="cov0" title="0">pluginActor := NewWasmPluginActor(pluginInstance)

                // 初始化插件
                if msg.Config != nil </span><span class="cov0" title="0">{
                        if err := pluginActor.HandleInitMessage(&amp;InitPluginMessage{Config: msg.Config}); err != nil </span><span class="cov0" title="0">{
                                a.RecordError(err)
                                // 继续加载其他插件，不中断
                                continue</span>
                        }
                }

                // 启动插件
                <span class="cov0" title="0">if err := pluginActor.OnStart(); err != nil </span><span class="cov0" title="0">{
                        a.RecordError(err)
                        // 继续加载其他插件，不中断
                        continue</span>
                }

                // A 添加到管理的插件列表
                <span class="cov0" title="0">a.plugins[pluginID] = pluginInstance
                a.pluginActors[pluginID] = pluginActor</span>
        }

        <span class="cov0" title="0">a.lastLoadTime = time.Now()
        return nil</span>
}

// HandleUnloadPlugin 处理卸载插件消息
func (a *WasmPluginManagerActor) HandleUnloadPlugin(msg *UnloadPluginMessage) error <span class="cov8" title="1">{
        a.mu.Lock()
        defer a.mu.Unlock()

        a.RecordMessageReceived()

        // 检查插件是否存在
        pluginActor, exists := a.pluginActors[msg.PluginID]
        if !exists </span><span class="cov8" title="1">{
                return fmt.Errorf("plugin %s not found", msg.PluginID)
        }</span>

        // 停止插件
        <span class="cov8" title="1">if err := pluginActor.OnStop(); err != nil </span><span class="cov0" title="0">{
                a.RecordError(err)
                // 继续卸载，不中断
        }</span>

        // 从管理列表中移除
        <span class="cov8" title="1">delete(a.plugins, msg.PluginID)
        delete(a.pluginActors, msg.PluginID)

        return nil</span>
}

// HandleListPlugins 处理列出插件消息
func (a *WasmPluginManagerActor) HandleListPlugins(msg *ListPluginsMessage) (*ListPluginsResponse, error) <span class="cov8" title="1">{
        a.mu.RLock()
        defer a.mu.RUnlock()

        a.RecordMessageReceived()

        var pluginInfos []plugin.PluginInfo
        for _, p := range a.plugins </span><span class="cov8" title="1">{
                info := p.GetInfo()
                pluginInfos = append(pluginInfos, info)
        }</span>

        <span class="cov8" title="1">return &amp;ListPluginsResponse{
                Plugins: pluginInfos,
        }, nil</span>
}

// HandleGetPlugin 处理获取插件消息
func (a *WasmPluginManagerActor) HandleGetPlugin(msg *GetPluginMessage) (*GetPluginResponse, error) <span class="cov8" title="1">{
        a.mu.RLock()
        defer a.mu.RUnlock()

        a.RecordMessageReceived()

        plugin, exists := a.plugins[msg.PluginID]
        if !exists </span><span class="cov8" title="1">{
                return &amp;GetPluginResponse{
                        Found: false,
                }, nil
        }</span>

        <span class="cov8" title="1">actor := a.pluginActors[msg.PluginID]
        info := plugin.GetInfo()

        return &amp;GetPluginResponse{
                Plugin:     plugin,
                PluginInfo: info,
                Actor:      actor,
                Found:      true,
        }, nil</span>
}

// GetMetrics 获取插件管理器度量指标
func (a *WasmPluginManagerActor) GetMetrics() map[string]interface{} <span class="cov8" title="1">{
        a.mu.RLock()
        defer a.mu.RUnlock()

        pluginCount := len(a.plugins)
        runningCount := 0

        for _, actor := range a.pluginActors </span><span class="cov8" title="1">{
                if actor.IsRunning() </span><span class="cov8" title="1">{
                        runningCount++
                }</span>
        }

        <span class="cov8" title="1">metrics := map[string]interface{}{
                "id":                "wasm_plugin_manager",
                "state":             string(a.State),
                "plugin_count":      pluginCount,
                "running_count":     runningCount,
                "messages_received": a.Metrics.MessagesReceived,
                "messages_sent":     a.Metrics.MessagesSent,
                "error_count":       a.Metrics.ErrorCount,
        }

        if a.Metrics.ErrorCount &gt; 0 </span><span class="cov0" title="0">{
                metrics["last_error"] = a.Metrics.LastError
        }</span>

        <span class="cov8" title="1">if !a.lastLoadTime.IsZero() </span><span class="cov0" title="0">{
                metrics["last_load_time"] = a.lastLoadTime.Format(time.RFC3339)
        }</span>

        <span class="cov8" title="1">return metrics</span>
}

// String 返回插件管理器Actor的字符串表示
func (a *WasmPluginManagerActor) String() string <span class="cov8" title="1">{
        metrics := a.GetMetrics()

        return fmt.Sprintf("WasmPluginManagerActor[state=%s, plugins=%d, running=%d]",
                a.State, metrics["plugin_count"], metrics["running_count"])
}</span>
</pre>
		
		</div>
	</body>
	<script>
	(function() {
		var files = document.getElementById('files');
		var visible;
		files.addEventListener('change', onChange, false);
		function select(part) {
			if (visible)
				visible.style.display = 'none';
			visible = document.getElementById(part);
			if (!visible)
				return;
			files.value = part;
			visible.style.display = 'block';
			location.hash = part;
		}
		function onChange() {
			select(files.value);
			window.scrollTo(0, 0);
		}
		if (location.hash != "") {
			select(location.hash.substr(1));
		}
		if (!visible) {
			select("file0");
		}
	})();
	</script>
</html>
